package com.g.ocache.impl.index;

import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import org.apache.commons.lang.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.g.ocache.enums.VisitingOrder;
import com.g.ocache.index.IndexBuilder;
import com.g.ocache.index.Leaf;
import com.g.ocache.index.Node;
import com.g.ocache.index.criterion.BasicCriterion;
import com.g.ocache.index.criterion.Criterion;
import com.g.ocache.index.criterion.OrderCriterion;
import com.g.ocache.index.criterion.RangedCriterion;


/**
 * 
 * @author zhangyf
 * @param <K>
 * @param <V>
 */
public class BasicIndex<K, V> extends AbstractIndex<K, V> {
	//
	private static final Logger LOGGER = LoggerFactory.getLogger(BasicIndex.class);
	
	//
	private final Node<K, V> root;
	
	/**
	 * 
	 */
	public BasicIndex(String name, IndexBuilder<K, V> builder) {
		this(name, builder, false);
	}
	
	public BasicIndex(String name, IndexBuilder<K, V> builder, boolean fair) {
		//
		super(name, builder, fair);
		
		//
		this.root = createNode(0, null, null);
	}
	
	/**
	 * 
	 */
	protected void doStart() throws Exception {
		// NOP
	}

	protected void doStop() throws Exception {
		// NOP
	}

	/**
	 * 
	 */
	public void clear() {
		lock(true);
		try {
			root.clear();
		} finally {
			unlock(true);
		}
	}

	public boolean add(K key, V value) {
		//
		Object[] ids = getBuilder().build(key, value);
		if(ids == null || ids.length != getBuilder().getMaxDepth()) {
			throw new IllegalArgumentException("invalid ids: " + ArrayUtils.toString(ids) + ", expected length: " + (getBuilder().getMaxDepth()));
		}
		
		//
		boolean r = false;
		lock(true);
		try {
			Leaf<K, V> leaf = getLeaf(ids, true);
			r = leaf.addKey(key);
		} finally {
			unlock(true);
		}
		return r;
	}

	public boolean remove(K key, V value) {
		//
		Object[] ids = getBuilder().build(key, value);
		if(ids == null || ids.length != getBuilder().getMaxDepth()) {
			throw new IllegalArgumentException("invalid ids: " + ArrayUtils.toString(ids) + ", expected length: " + (getBuilder().getMaxDepth()));
		}
		
		//
		boolean r = false;
		lock(true);
		try {
			Leaf<K, V> leaf = getLeaf(ids, false);
			if(leaf != null) {
				r = leaf.removeKey(key);
			}
		} finally {
			unlock(true);
		}
		
		//
		if(!r) {
			LOGGER.warn("failed to remove, index: {}, key: {}, value: {}", new Object[]{getName(), key, value});
		}
		return r;
	}
	
	public void visit(Node.Visitor<K, V> visitor, VisitingOrder order) {
		//
		if(visitor == null) {
			throw new IllegalArgumentException("invalid parameter visitor");
		}
		if(order == null) {
			throw new IllegalArgumentException("invalid parameter order");
		}
		
		//
		lock(true);
		try {
			root.visit(visitor, order);
		} finally {
			unlock(true);
		}
	}
	
	private Leaf<K, V> createLeaf(int depth, Object id, Node<K, V> parent) {
		return new BasicLeaf<K, V>(id, parent, getBuilder().getLeafComparator());
	}
	
	private Node<K, V> createNode(int depth, Object id, Node<K, V> parent) {
		return new BasicNode<K, V>(id, parent, getBuilder().getNodeComparator(depth));
	}
	
	private Leaf<K, V> getLeaf(Object[] nodeIds, boolean createNew) {
		//
		Node<K, V> node = root;
		for(int depth = 0; depth < nodeIds.length; depth++) {
			//
			Node<K, V> child = node.getChild(nodeIds[depth]);
			if(child != null) {
				node = child;
				continue;
			}
			
			//
			if(!createNew) {
				return null;
			} 

			//
			if(depth == nodeIds.length - 1) {
				child = createLeaf(depth + 1, nodeIds[depth], node);
			} else {
				child = createNode(depth + 1, nodeIds[depth], node);
			}
			node = child;
		}
		return (Leaf<K, V>)node;
	}
	
	/**
	 * 
	 */
	public int count(Criterion... criteria) {
		//
		if(criteria == null || criteria.length == 0) {
			throw new IllegalArgumentException("invalid parameter criteria");
		}
		
		//
		if(getLoader() != null) {
			Set<Node<K, V>> toLoad = null;
			lock(true);
			try {
				toLoad = getNodesToLoad(criteria);
			} finally {
				unlock(true);
			}
			if(toLoad != null && toLoad.size() > 0) {
				load(toLoad);
			}
		}
		
		//
		int r = 0;
		lock(false);
		try {
			for(Node<K, V> node : queryNodes(criteria)) {
				r += countKeys(node);
			}
		} finally {
			unlock(false);
		}
		return r;
	}
	
	public List<V> query(Criterion... criteria) {
		return query(0, 0, criteria);
	}

	public List<V> query(int from, Criterion... criteria) {
		return query(from, 0, criteria);
	}

	public List<V> query(int from, int to, Criterion... criteria) {
		//
		if(criteria == null || criteria.length == 0) {
			throw new IllegalArgumentException("invalid parameter criteria");
		}
		
		//
		if(getLoader() != null) {
			Set<Node<K, V>> toLoad = null;
			lock(true);
			try {
				toLoad = getNodesToLoad(criteria);
			} finally {
				unlock(true);
			}
			if(toLoad != null && toLoad.size() > 0) {
				load(toLoad);
			}
		}
		
		//
		List<K> keys = new LinkedList<K>();
		lock(false);
		try {
			for(Node<K, V> node : queryNodes(criteria)) {
				//
				Criterion lc = null; // Get leaf criterion
				if(criteria.length > getBuilder().getMaxDepth()) {
					lc = criteria[getBuilder().getMaxDepth()];
				}
				
				//
				keys.addAll(queryKeys(node, lc));
			}
		} finally {
			unlock(false);
		}
		
		//
		int index = 0;
		List<V> values = new LinkedList<V>();
		for(Iterator<K> iter = keys.iterator(); iter.hasNext(); ) {
			//
			K key = iter.next();
			V v = getCache().get(key);
			if(v == null) {
				continue;
			}
			
			//
			if (index >= from) {
				if (to > 0 && index >= to) {
					break;
				} else {
					values.add(v);
				}
			}
			index++;
		}
		return values;
	}
	
	private int countKeys(Node<K, V> node) {
		int r = 0;
		if(node instanceof Leaf) {
			r += ((Leaf<K, V>) node).getKeyCount();
		} else {
			for(Node<K, V> child : node.getChildren()) {
				r += countKeys(child);
			}
		}
		return r;
	}
	
	private List<K> queryKeys(Node<K, V> node, Criterion criterion) {
		List<K> keys = new LinkedList<K>();
		if(node instanceof Leaf) {
			//
			Leaf<K, V> leaf = ((Leaf<K, V>) node);
			
			//
			if(criterion == null) {
				keys.addAll(leaf.getKeys());
			} else if(criterion instanceof OrderCriterion) {
				OrderCriterion oc = (OrderCriterion)criterion;
				keys.addAll(leaf.getKeys(oc.getOrder()));	
			} else {
				throw new IllegalArgumentException("unsupported criterion: " + criterion);
			}
		} else {
			for(Node<K, V> child : node.getChildren()) {
				keys.addAll(queryKeys(child, criterion));
			}
		}
		return keys;
	}
	
	private List<Node<K, V>> queryNodes(Criterion... criteria) {
		//
		List<Node<K, V>> nodes = new LinkedList<Node<K, V>>();
		nodes.add(root);
		
		//
		for(int depth = 0; depth < getBuilder().getMaxDepth(); depth++) { // Nodes only
			//
			if(depth >= criteria.length) {
				break;
			}
			
			//
			List<Node<K, V>> temp = nodes;
			nodes = new LinkedList<Node<K, V>>();
			for(Node<K, V> node : temp) {
				//
				if(criteria[depth] == null) {
					nodes.addAll(node.getChildren());
				} else if(criteria[depth] instanceof RangedCriterion) {
					RangedCriterion rc = (RangedCriterion)criteria[depth];
					nodes.addAll(node.getChildren(rc.getFrom(), rc.isFromInclusive(), rc.getTo(), rc.isToInclusive(), rc.getOrder()));
				} else if(criteria[depth] instanceof OrderCriterion) {
					OrderCriterion oc = (OrderCriterion)criteria[depth];
					nodes.addAll(node.getChildren(null, true, null, true, oc.getOrder()));
				} else if(criteria[depth] instanceof BasicCriterion) {
					BasicCriterion bc = (BasicCriterion)criteria[depth];
					for(Object condition: bc.getConditions()) {
						Node<K, V> child = node.getChild(condition); 
						if(child != null) {
							nodes.add(child);
						}
					}
				} else {
					throw new IllegalArgumentException("assertion failed, unsupported criterion: " + criteria[depth]);
				}
			}
		}
		return nodes;
	}
	
	/**
	 * 
	 */
	private Set<Node<K, V>> getNodesToLoad(Criterion... criteria) {
		//
		Set<Node<K, V>> toLoad = new HashSet<Node<K, V>>();
		List<Node<K, V>> nodes = new LinkedList<Node<K, V>>();
		nodes.add(root);
		for(int depth = 0; depth < getBuilder().getMaxDepth(); depth++) { // Nodes only
			//
			if(depth >= criteria.length) {
				break;
			}
			
			//
			List<Node<K, V>> temp = nodes;
			nodes = new LinkedList<Node<K, V>>();
			for(Node<K, V> node : temp) {
				//
				if(criteria[depth] == null) {
					if(node.isInitialized()) {
						nodes.addAll(node.getChildren());
					} else if(node != root) {
						toLoad.add(node);						
					}
				} else if(criteria[depth] instanceof RangedCriterion) {
					if(node.isInitialized()) {
						RangedCriterion rc = (RangedCriterion)criteria[depth];
						nodes.addAll(node.getChildren(rc.getFrom(), rc.isFromInclusive(), rc.getTo(), rc.isToInclusive(), rc.getOrder()));
					} else if(node != root) {
						toLoad.add(node);
					}
				} else if(criteria[depth] instanceof OrderCriterion) {
					if(node.isInitialized()) {
						OrderCriterion oc = (OrderCriterion)criteria[depth];
						nodes.addAll(node.getChildren(null, true, null, true, oc.getOrder()));
					} else if(node != root) {
						toLoad.add(node);						
					}
				} else if(criteria[depth] instanceof BasicCriterion) {
					BasicCriterion bc = (BasicCriterion)criteria[depth];
					for(Object id: bc.getConditions()) {
						Node<K, V> child = node.getChild(id); 
						if(child != null && child.isInitialized()) {
							nodes.add(child);
						} else {
							toLoad.add(child != null ? child : this.createNode(depth + 1, id, node));
						}
					}
				} else {
					throw new IllegalArgumentException("assertion failed, unsupported criterion: " + criteria[depth]);
				}
			}
		}
		return toLoad;
	}
}
